package com.zty.service.impl;

import com.zty.BO.ServerInfo;
import com.zty.DO.FileChunkMeta;
import com.zty.DO.MetaFile;
import com.zty.DTO.CompleteChunkFileDTO;
import com.zty.DTO.FileMeta;
import com.zty.VO.BucketVO;
import com.zty.VO.FileVO;
import com.zty.config.MetaConfig;
import com.zty.errors.BusinessException;
import com.zty.errors.EnumMetaException;
import com.zty.service.DiscoveryService;
import com.zty.service.MetaService;
import com.zty.util.RequestUtil;
import com.zty.utils.FilenameGenerator;
import com.zty.utils.ServerSelector;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.stereotype.Service;

import javax.servlet.http.HttpServletRequest;
import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;

/**
 * @Author ZTY
 * @Description
 * @Date 22:51 2023-12-23
 **/

@Service
@Slf4j
public class MetaServiceImpl implements MetaService {

    private final FilenameGenerator filenameGenerator;
    private final HttpServletRequest httpServletRequest;
    private final MetaConfig metaConfig;
    private final MongoTemplate mongoTemplate;
    private final DiscoveryService discoveryService;
    private final ServerSelector serverSelector;

    public MetaServiceImpl(FilenameGenerator filenameGenerator, HttpServletRequest httpServletRequest,MongoTemplate mongoTemplate, MetaConfig metaConfig, DiscoveryService discoveryService, ServerSelector serverSelector) {
        this.filenameGenerator = filenameGenerator;
        this.httpServletRequest = httpServletRequest;
        this.mongoTemplate = mongoTemplate;
        this.metaConfig = metaConfig;
        this.discoveryService = discoveryService;
        this.serverSelector = serverSelector;
    }

    @Override
    public MetaFile generate(FileMeta fileMeta) {

        Long filesize = fileMeta.getFilesize();
        String extension = fileMeta.getExtension();
        String bucketName = fileMeta.getBucketName();

        //  将客户端ip地址取出
        String clientIpAdr = RequestUtil.getIpAddr(httpServletRequest);

        //  拿到每个分片的大小
        Integer chunkSize = metaConfig.getChunkSize();


        //  通过公式计算得出分片数
        int totalChunk = (int) Math.ceil(filesize * 1.0 / chunkSize);

        //  防止重复上传措施
        String filename = filenameGenerator.generate(fileMeta, clientIpAdr);
        //  从mongo中查询是否已存在相同的文件名
        MetaFile metaFile = mongoTemplate.findById(filename, MetaFile.class);
        //  若存在则直接返回回去
        if (Objects.nonNull(metaFile)){
            return metaFile;
        }

        List<FileChunkMeta> chunks = createChunks(filesize, extension, bucketName, chunkSize, totalChunk, filename);

        metaFile = new MetaFile();
        metaFile.setFilename(filename)
                .setExtension(extension)
                .setFilesize(filesize)
                .setBucketName(bucketName)
                .setTotalChunk(totalChunk)
                .setChunks(chunks)
                .setCompleted(false);

        mongoTemplate.insert(metaFile);



        return metaFile;
    }

    @Override
    public synchronized void completeChunk(CompleteChunkFileDTO completeChunkFileDTO) {
        String filename = completeChunkFileDTO.getFilename();
        //  根据文件名从mongo中将对应的文件信息拿出来
        MetaFile metaFile = mongoTemplate.findById(filename, MetaFile.class);

        //  为空说明文件不存在，抛异常
        if (Objects.isNull(metaFile)){
            throw new BusinessException(EnumMetaException.NOT_ENOUGH_CHUNK_SERVER);
        }

        AtomicBoolean completed = new AtomicBoolean(true);

        //` 若不为空，判断文件的每个分片是否成功上传
        metaFile.getChunks().forEach(c->{
            if (c.getChunkNo().equals(completeChunkFileDTO.getChunkNo()) &&
            c.getAddress().equals(completeChunkFileDTO.getAddress()) &&
            c.getSchema().equals(completeChunkFileDTO.getSchema())){
                //  修改分片状态
                c.setCompleted(true);
                c.setChunkMd5(completeChunkFileDTO.getMd5());
            }

            //  有一个分片的上传状态为false，整个文件的上传状态都为false
            if (!c.getCompleted()){
                completed.set(false);
            }
        });

        //  设置最终文件是否上传完成
        metaFile.setCompleted(completed.get());

        mongoTemplate.save(metaFile);
    }

    @Override
    public MetaFile meta(String bucketName, String filename) {

        //  从mongo中找元信息
        MetaFile metaFile = mongoTemplate.findById(filename, MetaFile.class);

        //  为空抛异常
        if (Objects.isNull(metaFile)){
            throw new BusinessException(EnumMetaException.META_FILE_NOT_FOUND);
        }

        //  由于分片的文件是冗余存储在服务器中的，所以要先检查某个分片对应的服务器有哪些是存活着的

        //  先检查在元服务器端注册的服务器有哪些是存活的
        List<ServerInfo> serverInfos = discoveryService.aliveServers();

        //  将地址取出，查看改地址中存有哪些分片
        //  此时的地址是不带schema的
        Set<String> addressSet = serverInfos.stream().map(serverInfo ->
                serverInfo.getHost() + ":" + serverInfo.getPort()
        ).collect(Collectors.toSet());

        //  将存在meta中的chunk信息都提取出来
        List<FileChunkMeta> chunks = metaFile.getChunks();

        chunks.forEach(c->{
            //  取出每个chunk所保存的地址
            String address = c.getAddress();

            //  不存在则表示该机器挂掉了，要降低该机器的权重，以后少使用该机器上的分片
            if (!addressSet.contains(address)){
                c.setWeight(c.getWeight()>0 ? c.getWeight() - 1 : 0);
            }else {
                //  分片可能曾经出现过失联，权重有所降低，当它活回来时要及时将权重更新回来，不然这个分片永远都不可用了
                c.setWeight(metaConfig.getChunkInstanceMaxWeight());
            }

            //  再次检查这个分片是否上传完成
            if (!c.getCompleted()){
                throw new BusinessException(EnumMetaException.CHUNK_FILE_NOT_UPLOADED);
            }
        });

        //  将权重的信息重新保存回mongo
        metaFile.setChunks(chunks);
        mongoTemplate.save(metaFile);

        //  分片是冗余存储的，不是每个服务器上的副本都要下载一遍
        List<FileChunkMeta> chunkMetas = chunks.stream()
                //  先根据分片序号进行分组（由于分片是冗余存储的，所以存在同分片序号不同存储地址的几个副本）
                .collect(Collectors.groupingBy(FileChunkMeta::getChunkNo))
                .values()
                .stream()
                .parallel()
                //  每一组只需筛选出一个副本即可
                .map(fileChunkMetas -> fileChunkMetas.stream()
                        .peek(c -> {
                            String address = c.getAddress();
                            //  判断
                            if (!addressSet.contains(address)) {
                                c.setWeight(c.getWeight() > 0 ? c.getWeight() - 1 : 0);
                            }
                        })
                        //  筛出权重最大的服务器里的副本
                        .max(Comparator.comparing(FileChunkMeta::getWeight))
                        //  存在根本没取到的可能性，直接去new一个新的出来
                        .orElse(new FileChunkMeta())
                )
                //  只取出权重大于0的成员
                .filter(e -> e.getWeight() > 0)
                .toList();

        //  检查分片是否完整
        //  所下载下来的分片数如果小于原本切分好的分片数，则表示分片存在丢失的情况，文件损坏
        if (chunkMetas.size() < metaFile.getTotalChunk()){
            //  判断哪个（哪些）分片丢失了

            //  将分片编号归为一个set，便可判断是哪个分片丢失了
            Set<Integer> chunkNoSet = chunkMetas.stream()
                    .map(FileChunkMeta::getChunkNo)
                    .collect(Collectors.toSet());

            //  记录丢失的分片的列表
            List<Integer> lossChunkNo = new ArrayList<>();

            for (int i = 0; i < metaFile.getTotalChunk(); i++) {
                if (!chunkNoSet.contains(i)){
                    //  不包含则表示分片丢失了，写入列表中
                    lossChunkNo.add(i);
                }
            }

            //  将丢失的分片信息打印出来
            log.warn("文件 {} 存在分片不可用或丢失现象：{}",filename,lossChunkNo);
            throw new BusinessException(EnumMetaException.NO_CHUNK_META_AVAILABLE);
        }

        metaFile.setChunks(chunkMetas);
        return metaFile;
    }

    @Override
    public List<BucketVO> files() {
        //  拿到所有meta信息
        List<MetaFile> allFilesMeta = mongoTemplate.findAll(MetaFile.class);

        //  将meta信息转换为FileVO和BucketVO
        //<桶名,属于该桶的文件的元数据列表>
        Map<String, List<MetaFile>> bucketMap = allFilesMeta.stream()
                //  将所有数据根据桶名称进行分组
                .collect(Collectors.groupingBy(MetaFile::getBucketName));


        return bucketMap.entrySet().stream().map(
                entry -> {
                    List<FileVO> fileVOList = entry.getValue()
                            .stream()
                            .map(metaFile -> new FileVO()
                                .setFilename(metaFile.getFilename())
                                .setExtension(metaFile.getExtension())
                                .setBucketName(metaFile.getBucketName())
                                .setFileSize(metaFile.getFilesize()))
                                .toList();

                    return new BucketVO()
                            .setBucketName(entry.getKey())
                            .setFiles(fileVOList);
                }
        ).toList();
    }

    @Override
    public void delete(String bucketName, String filename) {
        Query query = new Query();
        query.addCriteria(Criteria.where("filename").is(filename));
        mongoTemplate.remove(query, MetaFile.class);
    }

    @Override
    public List<FileChunkMeta> chunkInfo(String bucketName, String filename, Integer chunkNo) {
        //  filename我不希望它带后缀名
        if (filename.contains(".")){
            filename = filename.split("\\.")[0];
        }
        System.out.println("bucketName = " + bucketName);
        System.out.println("filename = " + filename);
        System.out.println("chunkNo = " + chunkNo);

        //  拿filename去mongo中查询数据
        MetaFile metaFile = mongoTemplate.findById(filename, MetaFile.class);



        //  为空抛异常
        if (Objects.isNull(metaFile)){
            throw new BusinessException(EnumMetaException.META_FILE_NOT_FOUND);
        }

        //  筛出chunkNo和我们所需要的编号一致的
        return metaFile.getChunks()
                .stream()
                .filter(chunk -> chunk.getChunkNo().equals(chunkNo))
                .collect(Collectors.toList());
    }


    /**
     * 创建文件分片的元数据
     * @param filesize  文件大小
     * @param extension 文件后缀名
     * @param bucketName  文件存储桶
     * @param chunkSize 分片大小
     * @param totalChunk  分片数
     * @param filename  文件名
     * @return  分片元数据列表
     */
    private List<FileChunkMeta> createChunks(Long filesize,
                                             String extension,
                                             String bucketName,
                                             Integer chunkSize,
                                             int totalChunk,
                                             String filename) {
        List<FileChunkMeta> chunks = new ArrayList<>();
        List<ServerInfo> aliveServers = discoveryService.aliveServers();

        //  如果没有存活的机器
        //  直接将异常抛出
        if (aliveServers.size() == 0){
            throw new BusinessException(EnumMetaException.NOT_ENOUGH_CHUNK_SERVER);
        }

        long start = 0;
        for (int i = 0; i < totalChunk; i++) {

            //  每个分片都要去计算当前分片大小

            //  默认分片大小
            long currentChunkSize = chunkSize;

            //  防止出现最后一个分片的分片大小不足默认分片大小的情况
            if (filesize < (long) (i + 1) * chunkSize){
                currentChunkSize = filesize - (long) i * chunkSize;
            }

            //  选择机器去存储
            //  这里使用随机策略
            List<ServerInfo> selectedServers = serverSelector.select(aliveServers,
                    metaConfig.getChunkInstanceCount());

            //  遍历选中的机器
            for (ServerInfo selectedServer : selectedServers) {
                //  取出机器的地址
                String address = selectedServer.getHost() + ":"
                        + selectedServer.getPort();

                FileChunkMeta fileChunkMeta = new FileChunkMeta();
                fileChunkMeta.setFilename(filename)
                        .setAddress(address)
                        .setChunkNo(i)
                        .setBucketName(bucketName)
                        .setChunkStart(start)
                        .setChunkSize((int) currentChunkSize)
                        .setExtension(extension)
                        .setSchema(selectedServer.getSchema())
                        .setWeight(metaConfig.getChunkInstanceMaxWeight());

                //  上传完才能知道MD5的信息，留在后面上传完后再补上
                chunks.add(fileChunkMeta);
            }
            //  每遍历一个分片，起始位置都要改变一下
            start += currentChunkSize;
        }
        return chunks;
    }
}
